""""
Case Type   : D库常用语法兼容-常用函数兼容4
Case Name   : 导出导入
Create At   : 2025-9-24
Owner       : hejiuhuan
Description :
    1.创建D库并切换至D库安装shark扩展
    2.执行对应函数测试
    3.使用gs_dump导出旧库数据
    4.创建新D库并切换至新D库安装shark扩展
    5.使用gs_restore导入旧库数据
    6.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.成功
    6.成功
History     :
    1、2025-09-24 创建
"""
import os
import unittest
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from yat.test import Node
from yat.test import macro
log = Logger()
primary_sh = CommonSH('PrimaryDbUser')

class Opengauss_Compatibility_D_Function_FOURTH_COMPATIBILITY_Case0013(unittest.TestCase):
    def setUp(self):
        log.info(f'--------{os.path.basename(__file__)} start--------')
        self.userNode = Node('PrimaryDbUser')
        self.constant = Constant()
        self.db_d = 'db_d'
        self.db_d_new = 'db_d_new'
        self.dbname = 'postgres'
    
    def Opengauss_Compatibility_D_Function_FOURTH_COMPATIBILITY_Case0013(self):
        text = '------------------step1: 创建D库并切换至D库安装shark扩展; expect: 成功------------------'
        log.info(text)
        msg = primary_sh.check_connection_status(self.constant.START_STATUS_MSG)
        self.assertTrue(msg)
        create_db = f'drop database if exists {self.db_d};create database {self.db_d} dbcompatibility \'D\';'
        msg_create_db = primary_sh.execut_db_sql(create_db, dbname=self.dbname)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS, msg_create_db, '创建D库失败')
        create_ext = 'create extension shark;'
        msg_create_ext = primary_sh.execut_db_sql(create_ext, dbname=self.db_d)
        self.assertIn(self.constant.create_extension_success, msg_create_ext, '创建shark扩展失败')

        text = '------------------step2: 执行对应函数测试; expect: 成功------------------'
        log.info(text)
        isnull = 'select isnull(1,2);'
        msg_isnull = primary_sh.execut_db_sql(isnull, dbname=self.db_d)
        log.info(msg_isnull)
        self.assertIn('1', msg_isnull, 'ISNULL函数测试失败')
        atn2 = 'select atn2(1,2);'
        msg_atn2 = primary_sh.execut_db_sql(atn2, dbname=self.db_d)
        log.info(msg_atn2)
        self.assertIn('.463647609000806', msg_atn2, 'ATN2函数测试失败')
        log10 = 'select log10(1);'
        msg_log10 = primary_sh.execut_db_sql(log10, dbname=self.db_d)
        log.info(msg_log10)
        self.assertIn('0', msg_log10, 'LOG10函数测试失败')
        datediff = 'select datediff(year,\'2025-01-01\',\'2026-01-02\');'
        msg_datediff = primary_sh.execut_db_sql(datediff, dbname=self.db_d)
        log.info(msg_datediff)
        self.assertIn('1', msg_datediff, 'DATEDIFF函数测试失败')
        datediff_big = 'select datediff_big(year,\'2025-01-01\',\'2026-01-02\');'
        msg_datediff_big = primary_sh.execut_db_sql(datediff_big, dbname=self.db_d)
        log.info(msg_datediff_big)
        self.assertIn('1', msg_datediff_big, 'DATEDIFF_BIG函数测试失败')
        charindex = 'select charindex(\'1\',\'abc123xyz\');'
        msg_charindex = primary_sh.execut_db_sql(charindex, dbname=self.db_d)
        log.info(msg_charindex)
        self.assertIn('4', msg_charindex, 'CHARINDEX函数测试失败')
        cast = 'select cast(9/2 as char(1));'
        msg_cast = primary_sh.execut_db_sql(cast, dbname=self.db_d)
        log.info(msg_cast)
        self.assertIn('4', msg_cast, 'CAST函数测试失败')
        try_cast = 'select try_cast(\'1.123456789\' as numeric(10,5));'
        msg_try_cast = primary_sh.execut_db_sql(try_cast, dbname=self.db_d)
        log.info(msg_try_cast)
        self.assertIn('1.12346', msg_try_cast, 'TRY_CAST函数测试失败')
        convert = 'select convert(varchar,123456);'
        msg_convert = primary_sh.execut_db_sql(convert, dbname=self.db_d)
        log.info(msg_convert)
        self.assertIn('123456', msg_convert, 'CONVERT函数测试失败')
        try_convert = 'select try_convert(varchar,123456);'
        msg_try_convert = primary_sh.execut_db_sql(try_convert, dbname=self.db_d)
        log.info(msg_try_convert)
        self.assertIn('123456', msg_try_convert, 'TRY_CONVERT函数测试失败')

        text = '------------------step3: 使用gs_dump导出旧库数据; expect: 成功------------------'
        log.info(text)
        dump = f'source {macro.DB_ENV_PATH};'\
               f'gs_dump {self.db_d} -p {macro.DB_PORT} -f {macro.DB_INSTANCE_PATH}/old_db_d.dump -F c'
        log.info(dump)
        res = self.userNode.sh(dump).result()
        log.info(res)
        self.assertIn(f'dump database {self.db_d} successfully', res, 'gs_dump导出旧库数据失败')

        text = '------------------step4: 创建新D库并切换至新D库安装shark扩展; expect: 成功------------------'
        log.info(text)
        create_db = f'drop database if exists {self.db_d_new};create database {self.db_d_new} dbcompatibility \'D\';'
        msg_create_db = primary_sh.execut_db_sql(create_db, dbname=self.dbname)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS, msg_create_db, '创建新D库失败')
        create_ext = 'create extension shark;'
        msg_create_ext = primary_sh.execut_db_sql(create_ext, dbname=self.db_d_new)
        self.assertIn(self.constant.create_extension_success, msg_create_ext, '新D库创建shark扩展失败')

        text = '------------------step5: 使用gs_restore导入旧库数据; expect: 成功------------------'
        log.info(text)
        restore = f'source {macro.DB_ENV_PATH};'\
                  f'gs_restore {macro.DB_INSTANCE_PATH}/old_db_d.dump -p {macro.DB_PORT} -d {self.db_d_new}'
        log.info(restore)
        res = self.userNode.sh(restore).result()
        log.info(res)
        self.assertIn(f'restore operation successful', res, 'gs_restore导入旧库数据失败')
        rm_dmp = f'rm -rf {macro.DB_INSTANCE_PATH}/old_db_d.dump'
        msg_rm = self.userNode.sh(rm_dmp).result()

    def tearDown(self):
        text = '------------------step6: 清理环境; expect: 成功------------------'
        log.info(text)
        drop_db = f'drop database if exists {self.db_d_new};'
        msg_drop_db = primary_sh.execut_db_sql(drop_db, dbname=self.dbname)
        log.info(msg_drop_db)
        self.assertIn('DROP DATABASE', msg_drop_db, '删除新库失败')
        drop_db = f'drop database if exists {self.db_d};'
        msg_drop_db = primary_sh.execut_db_sql(drop_db, dbname=self.dbname)
        log.info(msg_drop_db)
        self.assertIn('DROP DATABASE', msg_drop_db, '删除旧库失败')
        log.info(f'--------{os.path.basename(__file__)} end--------')